package com.bomb.disruptor.db;

import com.lmax.disruptor.*;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.concurrent.Executors;

public class InsertSql {

    static String sqlFilePath = "C:\\Users\\Waferslove\\Desktop\\dm.sql";//sql文件的路径

    public static void main(String[] args) throws Exception {
        int BUFFER_SIZE = 1024 * 1024 * 8;
        int THREAD_NUMBERS = 128;
        EventFactory<Sql> eventFactory = new EventFactory<Sql>() {
            public Sql newInstance() {
                return new Sql();
            }
        };

        RingBuffer<Sql> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);

        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        InsertHandler[] handlers = new InsertHandler[64];
        for (int i = 0; i < handlers.length; i++) {
            handlers[i] = new InsertHandler("insert" + i);
        }

        WorkerPool<Sql> workerPool = new WorkerPool<Sql>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handlers);

        workerPool.start(Executors.newFixedThreadPool(THREAD_NUMBERS));

        //下面这个生产8个数据
        File file = new File(sqlFilePath);
        BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(file), "UTF-8"));
        //替换
        String line = null;
        //以行为单位进行遍历
        long index = 0;
        while ((line = br.readLine()) != null) {
            index++;
            long seq = ringBuffer.next();
            Sql sql = ringBuffer.get(seq);
            sql.setIndex(index);
            sql.setInsertSql(line.replace(";",""));
            ringBuffer.publish(seq);
            System.out.println("丢入：" + index);
        }

        Thread.sleep(1000);
        workerPool.halt();
    }
}  